package TransfromTest

import SourceTest.Source01.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, StreamExecutionEnvironment}

object Transfrom01 {
  def main(args: Array[String]): Unit = {

    /**
     * StreamExecutionEnvironment
     * ExecutionEnvironment 使用的画，下面无法使用keyby
     */
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream1 = env.readTextFile("D:\\codes\\IDEA_prjects\\FlinkDemo\\data\\sensor.txt")

    //TODO map
    val value1 = stream1.map(x => x * 2)

    //TODO flatMap
    val value2 = stream1.flatMap(
      _.split(" ")
    )

    //TODO filter
    val value3 = stream1.filter(
      _ == 1
    )

    //TODO keyBy reduce
    val value4 = stream1.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong,
        dataArray(2).trim.toDouble)
    })
      .keyBy("id")
      .reduce((x, y) => SensorReading(x.id, x.timestamp + 1, y.timestamp))


    //TODO split select
    val stream2 = stream1
      .map(
        data => {
          val strings = data.split(",")
          SensorReading(strings(0), strings(1).toLong, strings(2).toDouble)
        }
      )
    val splitStream = stream2.split(
      seneorData => {
        if (seneorData.timestamp > 30) Seq("high") else Seq("low")
      }
    )
    val high: DataStream[SensorReading] = splitStream.select("high")
    val low: DataStream[SensorReading] = splitStream.select("low")
    val all: DataStream[SensorReading] = splitStream.select("high", "low")

    //TODO connect comap
    val warning: DataStream[(String, Double)] = high.map(sensorData => (sensorData.id, sensorData.temperature))
    val connected: ConnectedStreams[(String, Double), SensorReading] = warning.connect(low)
    val coMap: DataStream[Product with Serializable] = connected.map(
      warningData => (warningData._1, warningData._2, "warning"),
      lowData => (lowData.id, "healthy")
    )

    //TODO union
    val unionStream: DataStream[SensorReading] = high.union(low).union(all)
    unionStream.print()

    /**
     * Connect 与 Union 区别：
     * 1． Union 之前两个流的类型必须是一样，Connect 可以不一样，在之后的 coMap中再去调整成为一样的。
     * 2.  Connect 只能操作两个流，Union 可以操作多
     */
    env.execute()
  }

}
